#include "pool.h"
#include <mqueue.h>
#include <stdio.h>
#include <stdlib.h>
#include <pthread.h>
#include <sys/types.h>

#define MQ_NAME "/pool_mq"

static mqd_t mq = -1;

static pthread_t *executors = NULL;
static int executor_count = 0;

void *executor_func(void *argv)
{
    Task task;
    while (1)
    {
        int result = mq_receive(mq, (char *)&task, sizeof(Task), 0);
        if (result >= 0)
        {
            task.task_func(task.argv);
        }
    }
}

int pool_init(int count)
{
    // 初始化消息队列
    struct mq_attr attr;
    attr.mq_maxmsg = 10;
    attr.mq_msgsize = sizeof(Task);

    mq = mq_open(MQ_NAME, O_CREAT | O_RDWR, 0644, &attr);
    if (mq < 0)
    {
        perror("mq_open");
        return -1;
    }

    // 开启后台executor;
    executor_count = count;
    executors = malloc(executor_count * sizeof(pthread_t));
    if (!executors)
    {
        mq_unlink(MQ_NAME);
        perror("malloc");
        return -1;
    }

    for (int i = 0; i < executor_count; i++)
    {
        int result = pthread_create(executors + i, NULL, executor_func, NULL);
        if (result < 0)
        {
            perror("pthread_create");
            exit(EXIT_FAILURE);
        }
    }

    return 0;
}

int pool_registerTask(Task *task)
{
    int result = mq_send(mq, (char *)task, sizeof(Task), 0);
    return result;
}

void pool_close()
{
    for (int i = 0; i < executor_count; i++)
    {
        pthread_cancel(executors[i]);
        pthread_join(executors[i], NULL);
    }

    free(executors);
    mq_unlink(MQ_NAME);
}
